package com.aispeech.asr.apis.ssc.websocket.emitter;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.function.Consumer;

import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;

import com.aispeech.asr.apis.ssc.websocket.connection.EmitterManager;
import com.aispeech.asr.apis.ssc.websocket.connection.Message;

import lombok.ToString;
import reactor.core.publisher.FluxSink;

@ToString
public abstract class Emitter implements Consumer<FluxSink<WebSocketMessage>> {
    private final String id;
    private final EmitterType type;
    private final EmitterManager manager;
    private FluxSink<WebSocketMessage> sink;
    private EmitterState state;
    private WebSocketSession session;
    private long lastReceiveTime;
    
    private ArrayBlockingQueue<Message> receivedQueue = null;
    private List<Long> receivedList = Collections.synchronizedList(new ArrayList<Long>());
    private List<Long> clientAckList = Collections.synchronizedList(new ArrayList<Long>());

    public Emitter(WebSocketSession session, String id, EmitterType type, EmitterManager manager) {
    	this.session = session;
        this.id = id;
        this.type = type;
        this.manager = manager;
        this.state = EmitterState.CONNECTING;
        this.receivedQueue = new ArrayBlockingQueue<>(65536);
    }

    public String getId() {
        return id;
    }

    public EmitterType getType() {
        return type;
    }
    
    public EmitterState getState() {
        return state;
    }
    
    public WebSocketSession getSession() {
        return session;
    }

    public EmitterManager getManager() {
        return manager;
    }

    public FluxSink<WebSocketMessage> getSink() {
        return sink;
    }
    
    public ArrayBlockingQueue<Message> getReceivedQueue() {
        return receivedQueue;
    }
    
    public List<Long> getReceivedList() {
        return receivedList;
    }
    
    public List<Long> getClientAckList() {
        return clientAckList;
    }
    
    public void setSink(FluxSink<WebSocketMessage> sink) {
        this.sink = sink;
    }
    
    public void setState(EmitterState state) {
    	this.state = state;
    }
    
    public long getLastReceiveTime() {
		return lastReceiveTime;
	}
    
    public void setLastReceiveTime(long lastReceiveTime) {
		this.lastReceiveTime = lastReceiveTime;
	}

    public boolean isReady() {
    	return state == EmitterState.READY;
    }
   
    @Override
    public void accept(FluxSink<WebSocketMessage> sink) {
        this.sink = sink;
        this.state = EmitterState.READY;
    }
    
    
    
}
